package org.ws.task.schedule.dispatcher;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.exception.NotFoundNodeException;
import org.ws.task.item.ExecuteContext;
import org.ws.task.repository.LeaderContextRepository;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.TaskItem;

import java.util.List;

/**
 * 调度器实现，根据任务Item获取的Node执行调度
 */
public abstract class AbstractDispatcher implements Dispatcher {

    @Override
    public void dispatch(TaskItem item, String taskName) {
        // 根据taskName获取当前分配到的有效Nodes
        List<NodeMetadata> taskWorkerNodes = NodeMetadataRepository.getInstance().getActiveWorkerNodes(taskName);
        if(CollectionUtils.isNotEmpty(taskWorkerNodes)){
            // 根据策略获取待分发的Node
            List<NodeMetadata> workerNodes = getNode(item,taskWorkerNodes);
            if(CollectionUtils.isNotEmpty(workerNodes)){
                // 执行分发
                for (NodeMetadata workerNode : workerNodes) {
                    String id = workerNode.getId();
                    // 将item分发到Leader节点对应taskName的Context中
                    // 无论是pull或push都会直接从context中获取待处理item
                    ExecuteContext context = LeaderContextRepository.getInstance().getContext(taskName, id);
                    context.add(item);
                }
                return;
            }
        }
        throw new NotFoundNodeException("task["+taskName+"] can not fond worker node.");
    }

    /**
     * 获取当前Item能够发送到的Node
     * @param item
     * @param taskWorkerNodes
     * @return
     */
    protected abstract List<NodeMetadata> getNode(TaskItem item,List<NodeMetadata> taskWorkerNodes);
}
